package org.jpush.core;

import java.util.HashMap;
import java.util.Iterator;
import java.util.LinkedList;
import java.util.List;
import java.util.Map;

import org.jpush.packages.OutgoingPacket;
import org.jpush.utils.Pointer;

/**
 * A <code>Channel</code> will be used to maintain the
 * packets sent to multiple connections. The key of the
 * connection and will be stored.
 * A queue of <code>PacketInfo</code> will be maintained
 * as a history.
 * 
 * @author Lei Wang
 *
 */
public class Channel
{
	/**
	 * A linked list implementation to store a list of
	 * information of <code>OutgoingPacket</code>
	 * 
	 * @author Lei Wang
	 *
	 */
	public static class PacketInfo
	{
		public OutgoingPacket packet;

		// The key to identify the PacketInfo when removed
		// objectCategory will be checked.
		public String killKey;

		// 
		public int objectCategory;
		
		// The position in the queue
		public int queuePosition;

		private int refCount;
		
		// Mean it is in the removedList
		public boolean blnWaitingForRemoval;

		public PacketInfo previous;
		public PacketInfo next;
	    
	    public PacketInfo(OutgoingPacket pPacket, String killKey, int objectCategory, PacketInfo previous)
	    {
	        if (previous != null)
	        {
	        	this.previous.next = this;
	            this.previous = previous;
	            this.queuePosition = previous.queuePosition + 1;
	        }
	        else
	        {
	            this.previous = null;
	            this.queuePosition = 1;
	        }
	        this.next = null;
	        this.packet = pPacket;
	        this.objectCategory = objectCategory;
	        this.killKey = killKey;
	        this.refCount = 0;
	        this.blnWaitingForRemoval = false;
	    }
	    
	    public void incrementRefCount()
	    {
	    	this.refCount = this.refCount + 1;
	    }
	    
	    public void decrementRefCount()
	    {
	    	this.refCount = this.refCount - 1;
	    }
	    
	    public void setRefCount(int refCount)
	    {
	    	this.refCount = refCount;
	    }
	    
	    public int getRefCount()
	    {
	    	return this.refCount;
	    }
	    
	    public boolean hasNoRef()
	    {
	    	return this.refCount == 0;
	    }
	}

	private static final Integer CLIENT_VALUE = new Integer(0);
    private AbstractChannelManager manager;

    private String channelName;

    private int maxPacket;
    private boolean requireRegistration;

    // Packet which is removed but still being used will be stored here.
    private List<PacketInfo> removedList = new LinkedList<PacketInfo>();

    private PacketInfo head;
    private PacketInfo tail;
    private int nPackets;

    private Map<String, Integer> connectionMap = new HashMap<String, Integer>();

    private Object lock;
    public Channel(AbstractChannelManager pManager, String channelName, boolean requireRegistration)
    {
        this.manager = pManager;
        this.channelName = channelName;
        this.requireRegistration = requireRegistration;

        tail = null;
        head = null;

        lock = new Object();
        nPackets = 0;
    }

    /**
     * Remove a <code>PacketInfo</code> from the queue
     * If the packet is still being used, it will be 
     * moved to the removedList, otherwise, it will be
     * removed directly
     * 
     * @param packetInfo  the one to be removed
     */
    private void internalRemove(PacketInfo packetInfo)
    {
        if (packetInfo.next != null)
        {
            packetInfo.next.previous = packetInfo.previous;
        }

        if (packetInfo.previous != null)
        {
            packetInfo.previous.next = packetInfo.next;
        }

        if (packetInfo.equals(head))
        {
            head = packetInfo.next;
        }

        if (packetInfo.equals(tail))
        {
            tail = packetInfo.previous;
        }

        this.nPackets = this.nPackets - 1;

        if(packetInfo.hasNoRef())
        {
            manager.deleteOutgoingPacket(packetInfo.packet);
            
            // For GC
            packetInfo.packet = null;
            packetInfo.previous = null;
            packetInfo.next = null;
            packetInfo.killKey = null;
        }
        else
        {
            removedList.add(packetInfo);
            packetInfo.blnWaitingForRemoval = true;
        }
    }

    /**
     * How many packets are allowed to maintained in
     * this channel. If reaches this setting, the 
     * head of the queue will be removed. 
     * 
     * @param maxPacket
     */
    public void setMaxPackets(int maxPacket)
    {
        this.maxPacket = maxPacket;
    }

    /**
     * @see pushPacket(OutgoingPacket packet, String killKey, int objectCategory)
     * @param packet
     */
    public void pushPacket(OutgoingPacket packet)
    {
    	pushPacket(packet, "", 0);
    }

    /**
     * @see pushPacket(OutgoingPacket packet, String killKey, int objectCategory)
     * @param packet
     */
    public void pushPacket(OutgoingPacket packet, String killKey)
    {
    	pushPacket(packet, killKey, 0);
    }

    /**
     * Push a packet into the Queue. A PacketInfo will be
     * added
     * 
     * @param packet
     * @param killKey
     * @param objectCategory
     */
    public void pushPacket(OutgoingPacket packet, String killKey, int objectCategory)
    {
        synchronized(lock)
        {
            PacketInfo packetInfo = new PacketInfo(packet, killKey, objectCategory, tail);

            tail = packetInfo;
            if (head == null)
            {
                head = packetInfo;
            }

            this.nPackets = this.nPackets + 1;
            if (nPackets > maxPacket)
            {
                internalRemove(head);
            }
        }
    }

    public void removePacket()
    {
    	removePacket("", 0);
    }

    public void removePacket(String killKey)
    {
    	removePacket(killKey, 0);
    }

    public void removePacket(String killKey, int objectCategory)
    {
        synchronized(lock)
        {
            PacketInfo packetInfo = head;
            while (packetInfo != null && packetInfo.killKey.equals(killKey) && packetInfo.objectCategory != objectCategory)
            {
                packetInfo = packetInfo.next;
            }

            if (packetInfo != null)
            {
                internalRemove(packetInfo);
            }
        }
    }

    /**
     * Subscribe a Connection into this channel, later 
     * the packets sent by this connection will be
     * recorded
     * 
     * @param connectionKey  the key to identify a connection
     */
    public void subscribeConnection(String connectionKey)
    {
        synchronized(lock)
        {
            connectionMap.put(connectionKey, CLIENT_VALUE);
        }
    }


    /**
     * Unsubscribe a Connection into this channel, later 
     * the packets sent by this connection will no longer
     * be recorded.
     * 
     * @param connectionKey  the key to identify a connection
     */
    public void unsubscribeConnection(String connectionKey)
    {
        synchronized(lock)
        {
            connectionMap.remove(connectionKey);
        }
    }


    /**
     * @see disposePacket(PacketInfo packetInfo, String connectionKey, boolean blnSuccess)
     * 
     * @param packetInfo
     * @param connectionKey
     */
    public void disposePacket(PacketInfo packetInfo, String connectionKey)
    {
    	disposePacket(packetInfo, connectionKey, true);
    }

    /**
     * Decrement a reference of the packet if a packet can
     * be found, and if the packet is no longer needed, it
     * will be removed
     * 
     * @param packetInfo
     * @param connectionKey
     * @param blnSuccess
     */
    public void disposePacket(PacketInfo packetInfo, String connectionKey, boolean blnSuccess)
    {
        synchronized(lock)
        {
        	if(blnSuccess)
        	{
                connectionMap.put(connectionKey, new Integer(packetInfo.queuePosition));
        	}
        	
        	//In all cases decrement refCount :
            packetInfo.decrementRefCount();
            
            //Check if item is in internal garbage list
            //If found, remove it from the removedList
            if (packetInfo.blnWaitingForRemoval && packetInfo.hasNoRef())
            {
            	Iterator<PacketInfo> it = removedList.iterator();
            	while (it.hasNext())
            	{
            		if (it.next().equals(packetInfo))
            		{
            			it.remove();
            		}
            	}
            }
        }
    }

    public void disposeAllPackets()
    {
    	//Delete all packets in the queue.
        PacketInfo packetInfo = head;
    	while (packetInfo != null)
    	{
    		PacketInfo temp = packetInfo;
    		packetInfo = packetInfo.next;
    		manager.deleteOutgoingPacket(temp.packet);
    		
    		// For GC
    		temp.previous = null;
    		temp.next = null;
    		temp.packet = null;
    		temp.killKey = null;
    	}

		// For GC
    	head = null;
    	tail = null;
    }

    /**
     * Get the packet of the connection by the connectionKey
     * 
     * @param connectionKey
     * @param pPacketInfo
     * @return
     */
    public OutgoingPacket getNextPacket(String connectionKey, Pointer<PacketInfo> pPacketInfo)
    {
    	synchronized(lock)
        {
    		Integer value = connectionMap.get(connectionKey);
    		if (value == null && requireRegistration)
    			return null;
    		
    		int previouslyInsertedId = 0;
    		if (value != null)
    			previouslyInsertedId = value.intValue();

    		// Try to find the PacketInfo by the position value
    		// stored in the map
    		PacketInfo searchItem = head;
            while (searchItem != null && searchItem.queuePosition <= previouslyInsertedId)
            {
                searchItem = searchItem.next;
            }

            // If found, increment the reference and return the packet
            if (searchItem != null)
            {
                searchItem.incrementRefCount();
                pPacketInfo.setValue(searchItem);
                return searchItem.packet;
            }
            
            return null;
        }
    }

    /**
     * Get the name of this channel
     * 
     * @return The text of the name
     */
    public String getChannelName()
    {
        return channelName;
    }
}
